package com.surfilter.massdata.spark.task;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.Map.Entry;
import java.util.Set;
import java.util.regex.Pattern;

import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import scala.Tuple2;

import com.act.sparkanalyz.log.SysLog;
import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.surfilter.massdata.spark.bean.Dns_tmp;
import com.surfilter.massdata.spark.bean.Domain;
import com.surfilter.massdata.spark.bean.ExDomain;
import com.surfilter.massdata.spark.bean.ExDomainRegex;
import com.surfilter.massdata.spark.util.BaseRedisDao;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.ConfigUtil;
import com.surfilter.massdata.spark.util.StringUtil;
import com.surfilter.massdata.spark.util.URLUtil;
import com.surfilter.massdata.spark.util.ipcheck.IpCheckUtil;

public class EtlTask implements ISparkTask {
	String EX_DOMAIN_REDIS_KEY = "EXDOMAIN";
	transient BaseRedisDao redisDao;
	Map<String, ExDomainRegex> exDomainMap;
	final TreeMap<Long,Long> ipMap;
	public EtlTask (){
		redisDao =new BaseRedisDao();
		exDomainMap = redisDao.getAllExDomainMap(EX_DOMAIN_REDIS_KEY);
		ipMap = CommonUtils.getIpExtensionMap();
	}
	
	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,
			Map<String, String> commandMap) {
		List<OutQueueEntity> list = new ArrayList<OutQueueEntity>();
		
		try {
			// 读取原始数据
			DataFrame dns_raw = dataFrames.get("d_dns_raw");
			DataFrame dns = null;
			DataFrame dnsRes = null;
			DataFrame domain = null;
			DataFrame domainRes = null;
			JavaSparkContext jsc = JavaSparkContext.fromSparkContext(dns_raw.sqlContext().sparkContext());
			final Broadcast<Map<String, ExDomainRegex>> broadBswdMap =jsc.broadcast(exDomainMap);
			dns = genDnsDataFormat(dns_raw,broadBswdMap.getValue());
			// 输出domain数据
			domain=genDomainDataFormat(dns,ipMap);
			//domain.show();
			// 发现泛域名
			genExdomain(domain);
			//加上本次发现的泛域名，处理dns数据
			Map<String, ExDomainRegex> exDomainMap_2=redisDao.getAllExDomainMap(EX_DOMAIN_REDIS_KEY);
			System.out.println("exDomainMap:"+exDomainMap.size());
			System.out.println("exDomainMap_2:"+exDomainMap_2.size());
			final Broadcast<Map<String, ExDomainRegex>> broadBswdMap2 =jsc.broadcast(exDomainMap_2);
			dnsRes=genDns(dns,broadBswdMap2.getValue());
			//生成Domain数据
			domainRes=genDomainDataFormat(dnsRes,ipMap);
			//domainRes.show();
			// 输出dns数据
		    list.add(new OutQueueEntity("dns", dnsRes));
		    list.add(new OutQueueEntity("domain", domainRes));			
		} catch (Exception e) {
			e.printStackTrace();
		}
		return list;
	}

	private DataFrame genDns(DataFrame dns,
			Map<String, ExDomainRegex> exDomainMap2) {
		DataFrame dnsRes = null;
		try {
			PairFunction<Row, Object, Object> f;
			// map key:domain + "\t" + dip + "\t" + cip
			// value: visit_count+"\t"+ 时间戳
			// 泛域名替换
			JavaPairRDD<String, String> dnsinit = dns.toJavaRDD().mapToPair(
					new PairIP(exDomainMap2, 1));
			// reduce 计算 visit_count <value: 1+"\t"+ 时间戳>
			JavaPairRDD<String, String> dns_ret = dnsinit
					.reduceByKey(new org.apache.spark.api.java.function.Function2<String, String, String>() {
						public String call(String v1, String v2)
								throws Exception {
							String str1[] = StringUtils.split(v1,"\t");
							String str2[] = StringUtils.split(v2,"\t");
							int num1 = Integer.valueOf(str1[0]);
							int num2 = Integer.valueOf(str2[0]);
							int num = num1 + num2;
							long time1 = Long.valueOf(str1[1]);
							long time2 = Long.valueOf(str2[1]);
							// 取最早发现时间
							long time = (time1 < time2 ? time1 : time2);
							String value = num + "\t" + time;
							return value;
						}
					});
			JavaRDD<Dns_tmp> rd = dns_ret.mapPartitions(new flatMapDNS_());
			dnsRes = dns.sqlContext().createDataFrame(rd, Dns_tmp.class);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return dnsRes;
	}

	private void genExdomain(DataFrame domain) {

		try {
			SysLog.info("*****************genExdomain************************");

			// 转成 key:<dip1> value:<domaim>
			// key:<dip2> value:<domaim>
			JavaPairRDD<String, String> exd1 = domain.toJavaRDD()
					.flatMapToPair(new transIPKeyFloatMap());

			// countByKey
			Map<String, Object> exd_count = exd1.countByKey();
			//System.out.println("exd_count:"+exd_count.toString());
			Iterator it = exd_count.keySet().iterator();
			int judgeNum = Integer.valueOf(ConfigUtil.get(
					"surfilter.massdata.exdomain.judge.num", "200"));
			System.out.println("judgeNum:"+judgeNum);
			Map<String, Object> exdmap = new HashMap<String, Object>();
			// 过滤掉小于阀值的key
			while (it.hasNext()) {
				String key = it.next().toString();
				int num = Integer.valueOf(exd_count.get(key).toString());
				if (num > judgeNum) {
					exdmap.put(key, exd_count.get(key).toString());
				}
			}

			// 从数据集里过滤掉小于阀值的泛域名
			JavaPairRDD<String, String> exd2 = exd1.filter(new exdFilter_(
					exdmap));
			//System.out.println("exd2:"+exd2.collectAsMap());
			// key:<dip1> value:<domaim> 转为 key:<domaim> value:<dip1>
			JavaPairRDD<String, String> exd3 = exd2
					.mapToPair(new transDomainKeyMapToPair());
			//System.out.println("exd3:"+exd3.collectAsMap());
			// key:<domaim> value:<dip1,dip2>
			JavaPairRDD<String, String> exd4 = exd3.reduceByKey(new ipMerge());
			// 数据写入redis
			//System.out.println("exd4:"+exd4.collectAsMap());
			int maxLimit = Integer.valueOf(ConfigUtil.get(
					"surfilter.massdata.exdomain.max.limit", "10000"));			
			exd4.foreachPartition(new FlatMapExDomain(maxLimit));			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	/**
	 * 生成dns数据
	 */
	public DataFrame genDnsDataFormat(DataFrame dns_raw,Map<String, ExDomainRegex> exDomainMap) {
		String[] DIps = ConfigUtil.getArray("surfilter.114.ips");
		String[] Cips = ConfigUtil.getArray("surfilter.filter.cips");
		DataFrame dns = null;
		try {
			// 过滤：域名规则校验、Dip过滤、Cip过滤
			System.out.println("*******************dns_filter begin******************");
			 JavaRDD<Row> dns_filter = dns_raw.toJavaRDD()
			 .filter(new filterIp(DIps,Cips));
			 System.out.println("*******************dns_filter end******************");
			// map key:domain + "\t" + dip + "\t" + cip
			// value: 1+"\t"+ 时间戳
			//用顶级域名进行泛域名处理
			 System.out.println("*******************mapToPair 泛域名 begin******************");
			JavaPairRDD<String, String> dnsrdd =dns_filter.mapToPair(new PairIP(exDomainMap,0));
			 System.out.println("*******************mapToPair 泛域名 end******************");

			 
			 System.out.println("*******************reduceByKey  begin******************");
			// reduce 计算 visit_count  <value: 1+"\t"+ 时间戳>
			JavaPairRDD<String, String> dns_ret = dnsrdd
					.reduceByKey(new org.apache.spark.api.java.function.Function2<String, String, String>() {
						public String call(String v1, String v2)
								throws Exception {
							String str1[]=StringUtils.split(v1,"\t");
							String str2[]=StringUtils.split(v2,"\t");
							int num1= Integer.valueOf(str1[0]);
							int num2= Integer.valueOf(str2[0]);
							int num =num1+num2;
							long time1=Long.valueOf(str1[1]);
							long time2=Long.valueOf(str2[1]);
							//取最早发现时间
							long time= (time1<time2 ? time1:time2);
							String value =num+"\t"+time;
							return value;
						}
					});
			
			System.out.println("*******************reduceByKey 泛域名 end******************");
			JavaRDD<Dns_tmp> rd= dns_ret.mapPartitions(new flatMapDNS_());
			dns = dns_raw.sqlContext().createDataFrame(rd, Dns_tmp.class);

		} catch (Exception e) {
			e.printStackTrace();
		}
		//dns.show();
		return dns;
	}
	public DataFrame genDomainDataFormat(DataFrame dns,TreeMap<Long,Long> map) {
		//DIP省外IP过滤
		JavaRDD<Row> dns_filter;
		if(map!=null &&map.size()>0){
			dns_filter = dns.javaRDD().filter(new filterProvinceDIP(map));
		}else{
			dns_filter=dns.javaRDD();
		}		
//		System.out.println("dns_filter:"+dns_filter.count());
//		dns_filter=dns.javaRDD();
//		System.out.println("dns_filter:"+dns_filter.count());
		//Dns 数据 去掉列 CIP 
		//<domain dip cip dns_visit_count findtime> 转成 <domain dip : dns_visit_count findtime>
		JavaPairRDD<String, String> domain_1 = dns_filter.mapToPair(
				new PairFunction<Row,String, String>(){

					@Override
					public Tuple2<String, String> call(Row line) throws Exception {
						String domain=line.getAs("domain").toString();
						String dip=line.getAs("dip").toString();
						String visit_count=line.getAs("dns_visit_count").toString();
						String findtime=line.getAs("findtime").toString();
						String key=domain+"\t"+dip;
						String value=visit_count+"\t"+findtime;
						return new Tuple2(key, value);
					}
				});
		//printJavaPairRDD_S_I(domain_1);
		
		//dns_visit_count加和  <domain dip : dns_visit_count	findtime>
		JavaPairRDD<String, String> domain_2=domain_1
				.reduceByKey(new org.apache.spark.api.java.function.Function2<String, String, String>() {

					@Override
					public String call(String v1, String v2)
							throws Exception {
						String str1[]=StringUtils.split(v1,"\t");
						String str2[]=StringUtils.split(v2,"\t");
						int num1= Integer.valueOf(str1[0]);
						int num2= Integer.valueOf(str2[0]);
						int num =num1+num2;
						long time1=Long.valueOf(str1[1]);
						long time2=Long.valueOf(str2[1]);
						//取最早发现时间
						long time= (time1<time2 ? time1:time2);
						String value =num+"\t"+time;
						return value;
					}
				});
		//printJavaPairRDD_S_I(domain_2);
		
		//key:DOMAIN   value:DIP1	VISITSCOUNT1	FINDTIME
		JavaRDD<Tuple2<String, String>> domain_3=domain_2.map(new Function<Tuple2<String, String>,Tuple2<String, String>>(){
			@Override
			public Tuple2<String, String> call(Tuple2<String, String> v1)
					throws Exception {
				String[] key=StringUtils.split(v1._1(),"\t");
				String domain =key[0];
				String dip=key[1];
				String value=v1._2();
				return new Tuple2<String, String> (domain,dip+"\t"+value);
			}
		});
		//printJavaRDD(domain_3);
		
		//key:DOMAIN   value:DIP1	VISITSCOUNT1	FINDTIME
		JavaPairRDD<String, String> domain_pair =domain_3.mapToPair(new PairFunction<Tuple2<String, String> ,String, String>(){

			@Override
			public Tuple2<String, String> call(Tuple2<String, String> line) throws Exception {
				return new Tuple2<String, String> (line._1() ,line._2());
			}
			
		});
		//printJavaPairRDD(domain_pair);
		
		//<domain :dip	dns_visit_count	findtime> ==> <domaim : dip1|dip2	dns_visit_count1|dns_visit_count2	findtime>
		JavaPairRDD<String, String> domain_mult_ip= domain_pair.reduceByKey(new Function2<String, String, String>(){

			@Override
			public String call(String v1, String v2) throws Exception {
				//:DIP1	VISITSCOUNT1	FINDTIME
				String [] value_1=StringUtils.split(v1,"\t");
				String [] value_2=StringUtils.split(v2,"\t");
				long time1=Long.valueOf(value_1[2]);
				long time2=Long.valueOf(value_2[2]);
				//取最早发现时间
				long time= (time1<time2 ? time1:time2);
				String value =value_1[0]+"|"+value_2[0]+"\t"+value_1[1]+"|"+value_2[1]+"\t"+time;
				return value;
			}
			
		});
		//printJavaPairRDD(domain_mult_ip);
		
		JavaRDD<Domain> rd=domain_mult_ip.mapPartitions(new flatMapDomain());
		DataFrame domain_DF = dns.sqlContext().createDataFrame(rd, Domain.class);
		return domain_DF;
	}
	
	public void printJavaPairRDD(JavaPairRDD rdd){
		System.out.println("***************************JavaPairRDD begin print************************************rdd.count():"+rdd.count());
		List <Tuple2<String, String>> rdd_list= rdd.collect();
		for (Tuple2<String, String> l:rdd_list){
			System.out.println("l._1():	"+l._1().toString());
			System.out.println("l._2():	"+l._2());
		}
		System.out.println("***************************JavaPairRDD print end************************************");
	}
	public void printJavaPairRDD_S_I(JavaPairRDD rdd){
		System.out.println("***************************JavaPairRDD_S_I begin print************************************rdd.count():"+rdd.count());
		List <Tuple2<String, Integer>> rdd_list= rdd.collect();
		for (Tuple2<String, Integer> l:rdd_list){
			System.out.println("l._1():	"+l._1().toString());
			System.out.println("l._2():	"+l._2());
		}
		System.out.println("***************************JavaPairRDD_S_I print end************************************");
	}
	public void printJavaRDD(JavaRDD rdd){
		System.out.println("***************************JavaRDD begin print************************************rdd.count():"+rdd.count());
		List <Tuple2<String, String>> rdd_list= rdd.collect();
		for (Tuple2<String, String> l:rdd_list){
			System.out.println("l._1():	"+l._1().toString());
			System.out.println("l._2():	"+l._2());
		}
		System.out.println("***************************JavaRDD print end************************************");
	}
}
/**
 * 原始dns数据过滤：域名规则校验、Dip过滤、Cip过滤、顶级域名过滤
 */
class filterIp implements Function<Row, Boolean> {
	String[] DIps;
	String[] Cips;
	filterIp (String[] DIps,String[] Cips){
		this.DIps=DIps;
		this.Cips=Cips;
	}
	@Override
	public Boolean call(Row line) throws Exception {

		if (line.size() < 5) {
			return false;
		}
		String URL_PREFIX_HTTP = "http://";
		String DOMIAIN_REGX = "^([a-zA-Z0-9]([a-zA-Z0-9\\-]{0,61}[a-zA-Z0-9])?\\.)+[a-zA-Z]{2,6}$";
		Pattern domainPattern = Pattern.compile(DOMIAIN_REGX);
		String domain = line.getAs("domain").toString();
		String dip = line.getAs("dip").toString();
		String cip = line.getAs("cip").toString();
		
		
		String filterDIps="";
		String filterCips="";
        for(String ip:DIps){
        	filterDIps=filterDIps+ip+",";
        }
        for(String ip:Cips){
        	filterCips=filterCips+ip+",";
        }
        
		
		if (StringUtils.isBlank(domain) || StringUtils.isBlank(dip)
				|| StringUtils.isBlank(cip)) {
			return false;
		}
		// 域名合法性校验
		if (!domainPattern.matcher(domain).matches()) {
			return false;
		}
		// 目的IP 114IP过滤
		if (filterDIps.contains(dip)) {
			return false;
		}
		// 源IP过滤
		if (filterCips.contains(cip)) {
			return false;
		}

		String topDomain = URLUtil.getDomainName(URL_PREFIX_HTTP + domain);
		if (topDomain == null) {
			return false;
		}
		return true;
	}
}

class PairIP implements PairFunction<Row, String, String> {
	
	String URL_PREFIX_HTTP = "http://";
	String DOMIAIN_REGX ="^([a-zA-Z0-9]([a-zA-Z0-9\\-]{0,61}[a-zA-Z0-9])?\\.)+[a-zA-Z]{2,6}$";
	Pattern domainPattern = Pattern.compile(DOMIAIN_REGX);
	int flag=0;
	Map<String, ExDomainRegex> exDomainMap;
	public PairIP(Map<String, ExDomainRegex> exDomainMap,int flag) {
		this.exDomainMap=exDomainMap;
		this.flag=flag;
	}

	@Override
	public Tuple2<String, String> call(Row line) throws Exception {
		String domain = line.getAs("domain").toString();
		String dip = line.getAs("dip").toString();
		String cip = line.getAs("cip").toString();
		String dns_visit_count= line.getAs("dns_visit_count").toString();
		String findtime =line.getAs("findtime").toString();
 
		 // guoji.news.baidu.com
		 String topDomain = URLUtil.getDomainName(URL_PREFIX_HTTP + domain);
		 if (topDomain == null) {
		 return null;
		 }
		 topDomain = topDomain.toLowerCase();
		 String exDomain = domain;
		 //泛域名处理
		 try{
		 //缺陷【smcs_all_v1.0.0.0】 
		 domain = exdomainDeal(topDomain, dip,exDomainMap);
		 }catch(Exception e){
		 domain = null;
		 }
		
		
		 if (domain == null) {
		 domain = exDomain;
		 exDomain = null;
		 }
		String key = "";
		key = domain + "\t" + dip + "\t" + cip;
		String value ="";
		if(this.flag==0){
			//findtime 转成时间戳 2017-05-02 06:00:00
			SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//小写的mm表示的是分钟  
			java.util.Date date=sdf.parse(findtime); 
			long time=date.getTime()/1000;
			value =dns_visit_count+"\t"+time;
		}else{
			value =dns_visit_count+"\t"+findtime;
		}
		
		return new Tuple2(key, value);
	}

	/**
	 * 过滤掉泛域名-IP,将匹配到的泛域名-IP替换成对应的域名-IP
	 * 
	 * @param domain
	 * @param dips
	 * @return
	 */
	
	private String exdomainDeal(String topDomain, String dips,Map<String, ExDomainRegex> exDomainMap) {
		String EX_DOMAIN_DEFAULT_IP = "0.0.0.0";
		if (StringUtils.isBlank(topDomain) || StringUtils.isBlank(dips)) {
			return null;
		}
		List<String> ipList = new ArrayList<String>();
		ipList.addAll(Arrays.asList(StringUtils.split(dips,"\\|")));
		String exDomain = null;
		if (exDomainMap.containsKey(topDomain)) {
			ExDomainRegex exDomainRegex = exDomainMap.get(topDomain);
			Set<String> exIps = exDomainRegex.getIpSet();
			if (exIps.contains(EX_DOMAIN_DEFAULT_IP)) {
				exDomain = exDomainRegex.getNewName();
				return exDomain;
			} else if (intersectionSet(ipList, exIps)) {
				exDomain = exDomainRegex.getNewName();
				return exDomain;
			}
		}
		return exDomain;
	}

	private boolean intersectionSet(List<String> list, Set<String> setB) {
		list.retainAll(setB);
		return list.size() > 0;
	}
}


class filterProvinceDIP implements Function<Row, Boolean>{
	private TreeMap<Long,Long> ipMap;
	filterProvinceDIP(TreeMap<Long,Long> map){
		this.ipMap=map;
	}
	@Override
	public Boolean call(Row v1) throws Exception {
		String dip=v1.getAs("dip");
		boolean flag=CommonUtils.isExists(ipMap,dip);
		return flag;
	}
}

class filterProvinceCIP implements Function<Row, Boolean>{
	private TreeMap<Long,Long> ipMap;
	filterProvinceCIP(TreeMap<Long,Long> map){
		this.ipMap=map;
	}
	@Override
	public Boolean call(Row v1) throws Exception {
		String cip=v1.getAs("cip");
		boolean flag=CommonUtils.isExists(ipMap,cip);
		return flag;
	}
}
class flatMapDNS implements FlatMapFunction<Iterator<Row>, Dns_tmp>{

	@Override
	public List<Dns_tmp> call(Iterator<Row> t) throws Exception {
		List<Dns_tmp> list=new ArrayList<Dns_tmp>();
		while(t.hasNext()){
			Row row =t.next();
			Dns_tmp e = new Dns_tmp();
			e.setCip(row.getAs("cip").toString());
			e.setDip(row.getAs("dip").toString());
			e.setDns_visit_count(Long.valueOf(row.getAs("dns_visit_count").toString()));
			e.setDomain(row.getAs("domain").toString());
			e.setFindtime(Long.valueOf(row.getAs("findtime").toString()));
			list.add(e);
		}
		return list;
	}
}



class  flatMapDNS_ implements FlatMapFunction<Iterator<Tuple2<String, String>>, Dns_tmp> {

	@Override
	public Iterable<Dns_tmp> call(Iterator<Tuple2<String, String>> t)
			throws Exception {
		List<Dns_tmp> list=new ArrayList<Dns_tmp>();
		// map key:domain + "\t" + dip + "\t" + cip
					// value: 1+"\t"+ 时间戳
		while(t.hasNext()){
			Tuple2<String, String> tt =t.next();
			String [] key=StringUtils.split(tt._1(),"\t");
			String [] value=StringUtils.split(tt._2(),"\t");
			Dns_tmp e = new Dns_tmp();
			e.setDomain(key[0]);
			e.setDip(key[1]);
			e.setCip(key[2]);
			e.setDns_visit_count(Long.valueOf(value[0]));
			e.setFindtime(Long.valueOf(value[1]));
			list.add(e);
		}
		return list;
	}
}

class  flatMapDomain implements FlatMapFunction<Iterator<Tuple2<String, String>>, Domain>{

	@Override
	public Iterable<Domain> call(Iterator<Tuple2<String, String>> t)
			throws Exception {

		List<Domain> list=new ArrayList<Domain>();
		//<domaim : dip1|dip2	dns_visit_count1|dns_visit_count2	findtime>
		while(t.hasNext()){
			Tuple2<String, String> tt =t.next();
			String  key=tt._1();
			String [] value=StringUtils.split(tt._2(),"\t");
			Domain e = new Domain();
			e.setDomain(key);
			e.setDip(value[0]);
			e.setDns_visit_count(value[1]);
			e.setFindtime(Long.valueOf(value[2]));
			
			list.add(e);
		}
		return list;
	
	}
	
}


class transIPKeyFloatMap implements PairFlatMapFunction<Row, String, String> {
	String URL_PREFIX_HTTP = "http://";

	@Override
	public Iterable<Tuple2<String, String>> call(Row line) throws Exception {
		List<Tuple2<String, String>> results = new ArrayList<>();
		String domain = line.getAs("domain");
		String ips = line.getAs("dip");
		String[] ip_list = StringUtils.split(ips,"\\|");

		String topDomain = URLUtil.getDomainName(URL_PREFIX_HTTP + domain);
		topDomain = topDomain.toLowerCase();

		for (String ip : ip_list) {
			results.add(new Tuple2<String, String>(ip+"\t"+topDomain, "1"));
		}

		return (Iterable<Tuple2<String, String>>) results;
	}
}

class exdFilter_ implements Function<Tuple2<String, String>, Boolean> {
	Map<String, Object> exd_count = null;

	public exdFilter_(Map<String, Object> exd_count) {
		this.exd_count = exd_count;
	}

	@Override
	public Boolean call(Tuple2<String, String> v1) throws Exception {
		String key = v1._1();
		if (exd_count.containsKey(key)) {
			return true;
		} else {
			return false;
		}
	}

}

class transDomainKeyMapToPair implements
		PairFunction<Tuple2<String, String>, String, String> {

	@Override
	public Tuple2<String, String> call(Tuple2<String, String> t)
			throws Exception {
		String []key=StringUtils.split(t._1(),"\t");
		String ip = key[0];
		String domain = key[1];
		return new Tuple2<String, String>(domain, ip);
	}

}

class ipMerge  implements Function2<String, String, String>{

	@Override
	public String call(String v1, String v2) throws Exception {
		if(v1.contains(v2)){
			return v1;
		}else if(v2.contains(v1)){
			return v2;
		}else{
			return v1+","+v2;
		}
			
		
	}
	
}


class FlatMapExDomain  implements VoidFunction<Iterator<Tuple2<String, String>>> {
	int maxLimit=10000;
	FlatMapExDomain(int maxLimit){
		this.maxLimit=maxLimit;
	}
	@Override
	public void call(Iterator<Tuple2<String, String>> t)
			throws Exception {
		String EX_DOMAIN_REDIS_KEY = "EXDOMAIN";
		String EX_DOMAIN_DEFAULT_IP = "0.0.0.0";
		BaseRedisDao redisDao =new BaseRedisDao();
		ObjectMapper mapper=new ObjectMapper();
		ObjectWriter writer=mapper.writer();
		
		while(t.hasNext()){
			// key:<domaim> value:<dip1,dip2>
			Tuple2<String, String> tt=t.next();
			String domain =tt._1();
			String value=tt._2();

			ExDomain exDomain = new ExDomain();
			//当前数据找到的域名、ip
			String[] value_list = StringUtils.split(value,",");
			//redis 里存的域名、ip
			String redis_value = redisDao.hgetExDomain(EX_DOMAIN_REDIS_KEY, domain);
			Set  ipSet = new HashSet();
			String ips_new = "";
			if(redis_value!=null){
				ExDomain redis_domain = mapper.readValue(StringUtil.toLowerJson(redis_value),
						ExDomain.class);
				String redis_ips = redis_domain.getIps();
				String[] ip_list = StringUtils.split(redis_ips,",");
				for (String i : ip_list) {
					 ipSet.add(i.trim());
				}
			}
			if(ipSet.contains(EX_DOMAIN_DEFAULT_IP)){
				
			}else{
				for (String i : value_list) {
					 ipSet.add(i.trim());
				}

				if ( ipSet.size() > maxLimit) {
					ips_new = EX_DOMAIN_DEFAULT_IP;
				} else {
					Iterator iterator=ipSet.iterator();
					while(iterator.hasNext()){
						ips_new+=iterator.next();
						ips_new+=",";
					}
					ips_new = ips_new.substring(0, ips_new.length() - 1);
				}
				
				exDomain.setDomain("www." + domain);
				exDomain.setIps(ips_new);
				exDomain.setSuffix(domain);
				exDomain.setFindtime(new SimpleDateFormat("yyyyMMdd")
						.format(new Date()));
				System.out.println("redis:"+StringUtil.toUpperJson(writer.writeValueAsString(exDomain)));
				redisDao.hsetExDomain(EX_DOMAIN_REDIS_KEY, domain,
						StringUtil.toUpperJson(writer.writeValueAsString(exDomain)));
				
				
			}
		}
	}
}